package com.example.mqdemo.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * rabbitMQ配置
 */
@Slf4j
@Configuration
public class RabbitConfig {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void init() {
        // 解决msg.getMessageProperties().getCorrelationId()获取为null
        rabbitTemplate.setCorrelationDataPostProcessor((message, correlationData) -> {
            MessageProperties messageProperties = message.getMessageProperties();
            String correlationId = correlationData.getId();
            messageProperties.setCorrelationId(correlationId);
            // 可以设置持久化，但与本文无关，因此没有附上
            return correlationData;
        });
        // 数据转换为json存入消息队列
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        // 消息是否成功发送到Exchange
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息:{}成功发送到Exchange", correlationData);
            } else {
                log.warn("消息发送到Exchange失败:{}", cause);
            }
        });

        // 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
        rabbitTemplate.setReturnsCallback(message -> {
            log.info("Exchange路由到Queue失败，返回消息回调:{} 应答代码:{} 回复文本:{} 交换器:{} 路由键:{}",
                    message.getMessage(), message.getReplyCode(), message.getReplyText(), message.getExchange(), message.getRoutingKey());
        });
    }

    /**
     * 解决msg.getMessageProperties().getCorrelationId()获取为null
     *
     * @return
     */
    @Bean
    public MessagePostProcessor correlationIdProcessor() {
        return new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message, Correlation correlation) {
                MessageProperties messageProperties = message.getMessageProperties();

                if (correlation instanceof CorrelationData) {
                    String correlationId = ((CorrelationData) correlation).getId();
                    messageProperties.setCorrelationId(correlationId);
                }
                // 可以设置持久化，但与本文无关，因此没有附上
                return message;
            }

            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                return message;
            }
        };
    }

}
